package com.ecreditpal.audio.netty.handler;

import com.ecreditpal.audio.AsrCallback;
import com.ecreditpal.audio.AsrRunner;
import com.ecreditpal.audio.model.AudioModel;
import com.ecreditpal.audio.model.AudioSession;
import com.ecreditpal.audio.platform.PlatformBalancer;
import com.ecreditpal.audio.rpc.client.RpcSpeechClientService;
import com.ecreditpal.audio.service.SeechRecordService;
import com.ecreditpal.audio.util.BaseUtil;
import com.ecreditpal.audio.wav.Wav;
import io.netty.channel.ChannelHandlerContext;
import lombok.extern.slf4j.Slf4j;

import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @author lifeng
 * @version 1.0 on 2018/7/12.
 *          音频识别处理任务,因为有阻塞队列,内部还需要将任务交给线程池
 */
@Slf4j
public class AudioProcessor implements Runnable {
    private AudioSession session;
    private RpcSpeechClientService speechService;
    private SeechRecordService seechRecordService;
    private ChannelHandlerContext ctx;
    private PlatformBalancer balancer;

    public AudioProcessor(AudioSession audioSession, RpcSpeechClientService speechService, SeechRecordService seechRecordService,
                          ChannelHandlerContext ctx, PlatformBalancer balancer) {
        this.session = audioSession;
        this.speechService = speechService;
        this.seechRecordService = seechRecordService;
        this.ctx = ctx;
        this.balancer = balancer;
    }

    /**
     * When an object implementing interface <code>Runnable</code> is used
     * to create a thread, starting the thread causes the object's
     * <code>run</code> method to be called in that separately executing
     * thread.
     * <p>
     * The general contract of the method <code>run</code> is that it may
     * take any action whatsoever.
     *
     * @see Thread#run()
     */
    @Override
    public void run() {
        try {
            while (true) {
                if (!ctx.channel().isActive()){
                    log.info("channel not active,close the task");
                    break;
                }
                AudioModel.Audio req = session.getQueue().poll();

                if (req == null) {
                    try {
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) {
                        log.error("can not sleep for 1s", e);
                    }
                    log.info("req is null");
                    continue;
                } else {
                    log.info("task index {}", req.getIndex());
                }

                session.setCaseId(req.getCaseid());
                session.setCallId(req.getCallid());

                //读取数据
                ByteBuffer bf = req.getData().asReadOnlyByteBuffer();
                log.info("get buffer");
                if (bf.hasRemaining()) {
                    byte[] newMsg = new byte[bf.remaining()];
                    bf.get(newMsg);
                    log.info("get buffer success,length:{}", newMsg.length);

                    //合并历史数据
                    byte[] data = mergeData(newMsg, req.getType(), session);

                    List<Double> points;
                    points = speechService.sendMessage(data);


                    log.info("cut size: {}", points.size());
                    double offset = 0.0;
                    if (points.size() > 0) {
                        for (Double p : points) {
                            AudioModel.Text.Builder builder = AudioModel.Text.newBuilder();
                            builder.setCaseid(req.getCaseid());
                            builder.setCallid(req.getCallid());
                            builder.setType(req.getType());
                            builder.setBeginTime((int) ((session.getSessionOffset(req.getType()) == 0 ? offset : session.getSessionOffset(req.getType()) + offset) * 1000));
                            builder.setEndTime((int) ((session.getSessionOffset(req.getType()) + p) * 1000));

                            log.info("start:{},end:{}", builder.getBeginTime(), builder.getEndTime());

                            AsrCallback asrCallback = new AsrCallback(ctx, session, builder);
                            byte[] chunk = Wav.cutBody(data, offset, p, session.getByteRate());

                            AsrRunner asrRunner;
                            if (session.isLongChunk(req.getType())) {
                                asrRunner = balancer.getLongRunner();
                                session.setLongChunk(req.getType(), false);
                            } else if (p - offset > 5) {
                                asrRunner = balancer.getLongRunner();
                            } else {
                                asrRunner = balancer.chooseRunner();
                            }
                            asrRunner.process(chunk, asrCallback);

                            //更新最近一个分割的节点
                            offset = p;
                        }
                        session.setSessionOffset(req.getType(), session.getSessionOffset(req.getType()) + offset);
                        cutData(req.getType(), offset);
                    } else {
                        session.setLongChunk(req.getType(), true);
                    }

                    if (req.getIsLast()) {
                        log.info("it is the last chunk,index: {}", req.getIndex());
                        AsrRunner asrRunner = session.isLongChunk(req.getType()) ? balancer.getLongRunner() : balancer.chooseRunner();
                        AudioModel.Text.Builder builder = AudioModel.Text.newBuilder();
                        builder.setCaseid(req.getCaseid());
                        builder.setCallid(req.getCallid());
                        builder.setType(req.getType());
                        builder.setBeginTime((int) (session.getSessionOffset(req.getType()) * 1000));
                        builder.setIsLast(true);

                        session.setSessionOffset(req.getType(), session.getSessionOffset(req.getType()) +
                                Wav.getWavLengthNoBody(session.getLeftByType(req.getType()), session.getByteRate()));
                        builder.setEndTime((int) (session.getSessionOffset(req.getType()) * 1000));
                        AsrCallback asrCallback = new AsrCallback(ctx, session, builder);
                        asrRunner.process(session.getLeftByType(req.getType()), asrCallback);

                        try {
                            seechRecordService.recordToRemote(session);
                        } catch (Exception e) {
                            log.error("can not record audio text to remote", e);
                        }

                        session.clearSession();
                        break;
                    }
                }
            }
        } catch (Exception e) {
            log.error("error occur while do recognize", e);
        }
    }


    /**
     * 语音流不是一收到就会进行识别,而是将其进行断句处理.
     * 断句剩下的部分视为不完整的部分,等待下次语音流到来时一起处理.
     * 所以新语音流到来时,会对语音流进行合并处理
     *
     * @param newMsg       新语音流
     * @param audioType    语音的类型,local or remote
     * @param audioSession 每一次通话的会话信息
     * @return 合并后的语音流
     */
    public byte[] mergeData(byte[] newMsg, AudioModel.AudioType audioType, AudioSession audioSession) {
        byte[] data;
        if (audioType == AudioModel.AudioType.LOCAL) {
            audioSession.setLocalLeft(audioSession.getLocalLeft() == null ?
                    newMsg : BaseUtil.mergeBytes(audioSession.getLocalLeft(), newMsg));
            data = audioSession.getLocalLeft();
        } else {
            audioSession.setRemoteLeft(audioSession.getRemoteLeft() == null ?
                    newMsg : BaseUtil.mergeBytes(audioSession.getRemoteLeft(), newMsg));
            data = audioSession.getRemoteLeft();
        }
        return data;
    }

    /**
     * 得到断句的时间节点,对语音流进行分割处理
     *
     * @param audioType 语音流类型 local or remote
     * @param offset    分割的节点
     */
    public void cutData(AudioModel.AudioType audioType, double offset) {
        if (audioType == AudioModel.AudioType.LOCAL) {
            session.setLocalLeft(Wav.cutSingle(session.getLocalLeft(), offset, session.getByteRate()));
        } else {
            session.setRemoteLeft(Wav.cutSingle(session.getRemoteLeft(), offset, session.getByteRate()));
        }
    }
}
